package com.uatair.common.service.impl;

import com.uatair.common.kafka.domain.protobuf.FlightProto;
import com.uatair.common.mqtt.util.ProtobufUtil;
import com.uatair.common.service.DroneDataService;
import com.google.protobuf.InvalidProtocolBufferException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.io.IOException;

@Service
public class DroneDataServiceImpl implements DroneDataService {

    private static final Logger logger = LoggerFactory.getLogger(DroneDataServiceImpl.class);

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final String topic = "lianhefeiji-fly-tohenan";

    // 构造器注入
    public DroneDataServiceImpl(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    public boolean sendFlightRecord(FlightProto.FlightRecord flightRecord) {
        try {
            String massage = ProtobufUtil.toJson(flightRecord);

            // 异步发送带回调
            ListenableFuture<SendResult<String, String>> future =
                    kafkaTemplate.send(topic, massage);

            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    logger.info("成功发送飞行记录到Kafka, UUID: {}, 分区: {}",
                            flightRecord.getUuid(),
                            result.getRecordMetadata().partition());
                }

                @Override
                public void onFailure(Throwable ex) {
                    logger.error("飞行记录发送失败, UUID: {}", flightRecord.getUuid(), ex);
                }
            });

            return true;
        } catch (Exception e) {
            logger.error("飞行记录序列化失败: {}", e.getMessage(), e);
            return false;
        }
    }

    @Override
    public void processFlightRecord(String recordBytes) {
        try {
            FlightProto.FlightRecord record = ProtobufUtil.fromJson(recordBytes);

            logger.info("接收到飞行记录 - UUID: {}, 机型: {}, 位置: {}/{}",
                    record.getUuid(),
                    record.getDroneType(),
                    record.getPosition().getLatitude(),
                    record.getPosition().getLongitude());

            // TODO: 实际业务处理逻辑
            // 1. 数据持久化
            // 2. 业务分析
            // 3. 触发后续流程

        } catch (InvalidProtocolBufferException e) {
            logger.error("飞行记录解析失败: {}", e.getMessage(), e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
